/**
 * 
 */
package PagedFile;

import java.nio.channels.FileChannel;

/**
 * @author Neil
 * Buffer Manager manages the buffer pool in the memory.
 * It is only responsible for the allocation and free of pages.
 * It doesn't care if the buffer has non-valid data, or even has no data.
 */
public class BF_Manager {
	public static final int PF_BUFFER_SIZE=5000;
	public static final int WAITING_TIME=1000;
	
	Buffer[] bufferpool;
	int available;
	byte[] zeros = new byte[Buffer.BLOCK_SIZE];
	
	public BF_Manager() {
		bufferpool = new Buffer[PF_BUFFER_SIZE];
		for (int i=0; i<PF_BUFFER_SIZE; ++i)
			bufferpool[i] = new Buffer();
		available = PF_BUFFER_SIZE;
	}
	
	public int available() {
		return available;
	}
	
	public Buffer find(Block blk){
		for (Buffer buf : bufferpool)
			if (buf.block != null && buf.block.filename == blk.filename && buf.block.pageNum == blk.pageNum )
				return buf;
		return null;
	}
	
	public synchronized Buffer pin(Block blk, FileChannel fc){
		try {
			long timestamp = System.currentTimeMillis();
			Buffer buff = nonBlockPin(blk, fc);
			while (buff == null && !waitingTooLong(timestamp)) {
				wait(WAITING_TIME);
				buff = nonBlockPin(blk, fc);
			}
			if (buff == null)
				throw new PF_Abort_Exception("Buffer full");
			return buff;
		} catch(InterruptedException e) {
			throw new RuntimeException("Interrupted");
		}
	}
	
	public synchronized void unpin(Buffer buf) {
		buf.unpin();
		buf.timeStamp = System.currentTimeMillis();
		if (!buf.isPinned()) {
			++available;
			notifyAll();
		}
	}

	private Buffer nonBlockPin(Block blk, FileChannel fc){
		Buffer p = findExistingPage(blk, fc);
		if (p==null) {
			p = chooseUnpinnedPage();
			if (p == null)
				return null;
			if (p.isModified()) {
				if (p.fc == null)
					throw new PF_Abort_Exception("Fatal");
				p.write();
			}
			p.contents.clear();
			p.contents.put(zeros);
			p.contents.clear();
			p.block = blk;
			p.fc = fc;
			p.modifiedBy=-1;
			p.pinCount=0;
			p.read = false;
		}
		if (!p.isPinned())
			--available;
		p.pin();
		return p;
	}

	private Buffer findExistingPage(Block blk, FileChannel fc) {
		for (Buffer p : bufferpool) {
			Block b = p.block;
			if (b != null && b.filename == blk.filename && b.pageNum == blk.pageNum) {
				if (p.fc != fc)
					throw new PF_Abort_Exception("Fatal");
				return p;
			}
		}
		return null;
	}

	private long Cal(Buffer b) {
		return ((b.modifiedBy>0?1:0)<<20 + b.timeStamp>>10);
	}
	private Buffer chooseUnpinnedPage() {
		Buffer pinnedBuffer = null;
		long cal = Long.MAX_VALUE;
		for (Buffer p : bufferpool) {
			if ( !p.isPinned() && Cal(p)<cal ) {
				pinnedBuffer = p;
				cal = Cal(p);
			}
		}
		return pinnedBuffer;
	}

	private boolean waitingTooLong(long starttime) {
		return System.currentTimeMillis() - starttime > WAITING_TIME*2;
	}

	public void unpinAll(FileChannel fc) {
		for (Buffer buf : bufferpool)
			while ( buf.isPinned() && buf.fc == fc ) 
				unpin(buf);
	}
	
	public void flushAll(FileChannel fc) {
		for (Buffer buf : bufferpool)
			if ( buf.fc == fc )  
				buf.write();
	}
}
